package com.aotain.project.apollo.bolt;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

public class UrlAbnorBolt extends AbstractBolt{


	private static final long serialVersionUID = 1L;
	private String suffixname = null;
	private String domainlist = null;
	
	HashMap<String, HashMap<String, Long>> hmpath = new HashMap<String, HashMap<String, Long>>();
	HashMap<String, HashMap<String, Long>> hmpara = new HashMap<String, HashMap<String, Long>>();
	HashMap<String,HashMap<Integer,Integer>> domainlen = new HashMap<String,HashMap<Integer,Integer>>();
	HashMap<String,String> hmlen = new HashMap<String,String>();
	HashMap<String, String> hmproba = new HashMap<String, String>();
	String statConfig = null; 
	String probaConfig = null; 
	
	static Configuration conf=new Configuration();  
	
	public UrlAbnorBolt(String staturi,String probauri)
	{
		statConfig = staturi;
		probaConfig = probauri;
	}
			
	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {

		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour","abntype"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

	@Override
	public void cleanup() {
		
	}


	
	public void getStatConfigFiles() throws IOException{  
		
		FileSystem fshdfs = FileSystem.get(URI.create(statConfig),conf);
        FSDataInputStream in = null;
        BufferedReader bis = null;
        Path path = new Path(statConfig);
        try{	        
	        if(fshdfs.exists(path) && fshdfs.isDirectory(path)){
	    		List<Path> listpa = new ArrayList<Path>();
				for(FileStatus file:fshdfs.listStatus(path)){					
	       			if(file.isDirectory()){
	       				for(FileStatus fsp:fshdfs.listStatus(file.getPath())){
	       					if(fsp.isFile()) listpa.add(fsp.getPath());
	        			}
	       				
	       			}else if(file.isFile()){
	       				listpa.add(file.getPath());
	       			}
				}				
				for(Path hdfsfile : listpa){
					in = fshdfs.open(hdfsfile);
		           	bis = new BufferedReader(new InputStreamReader(in,"UTF8")); 
		           	String line = "";
		           	while ((line = bis.readLine()) != null) {  
		           		String items[] = line.split("\\|",-1);
		           		if(items.length>3){
			           		switch(items[1]){
			           		case "0":
			           			if(!hmlen.containsKey(items[0])){
			           				hmlen.put(items[0], items[2]+"|"+items[3]);
			           			}
			           			break;
			           		case "1":		           			
			           			HashMap<String, Long> pps = null;	
		    		        	if(hmpath!=null && hmpath.containsKey(items[0])){	    		        		
		    		        		pps = hmpath.get(items[0]);
		    		        		pps.put(items[2],Long.parseLong(items[3].trim()));	   
		    		        		
		    		        	}else{
		    		        		pps = new HashMap<String, Long>();
		    		        		pps.put(items[2],Long.parseLong(items[3].trim()));
		    		        		hmpath.put(items[0], pps);
		    		        	}
			           			break;
//			           		case "2":		           			
//			           			HashMap<String, Long> pas = null;	
//		    		        	if(hmpara!=null && hmpara.containsKey(items[0])){
//		    		        		pas = hmpara.get(items[0]);	    		        		
//		    		        		pas.put(items[2],Long.parseLong(items[3].trim()));
//		    		        	}else{
//				        			pas = new HashMap<String, Long>();
//				        			pas.put(items[2],Long.parseLong(items[3].trim()));
//				        			hmpara.put(items[0], pas);
//				        		}
//			           			break;
			           		}
		           		}else{
		           			Logger.getRootLogger().info("statConfig line format error>>>>>>>>>>>>>>>>"+line);
		           		}
		           	}
				}
	        }
	        
	        path = new Path(probaConfig);
	        if(fshdfs.exists(path) && fshdfs.isDirectory(path)){	        		
	        		List<Path> listproba = new ArrayList<Path>();
	    			for(FileStatus file:fshdfs.listStatus(path)){
	    				
	           			if(file.isDirectory()){
	           				for(FileStatus fsp:fshdfs.listStatus(file.getPath())){
	           					if(fsp.isFile()) listproba.add(fsp.getPath());
	            			}
	           				
	           			}else if(file.isFile()){
	           				listproba.add(file.getPath());
	           			}
	    			}
	    			for(Path hdfsfile : listproba){
	    				in = fshdfs.open(hdfsfile);
	    	           	bis = new BufferedReader(new InputStreamReader(in,"UTF8")); 
	    	           	String line = "";
	    	           	while ((line = bis.readLine()) != null) {
	    	           		String items[] = line.split("\\|",-1);
	    	           		if(items.length == 3){
		    	           		if(!hmproba.containsKey(items[0])){
		    	           			hmproba.put(items[0], items[1]+"|"+items[2]);
		    	           		}
		    	           	}else{
			           			Logger.getRootLogger().info("probaConfig line format error>>>>>>>>>>>>>>>>"+line);
			           		}
	    	           	}
	    			}
	        }	        
        }catch(Exception ex){
    		Logger.getRootLogger().error("UrlAbnorBolt_getStatConfigFiles error >>>>>>>>>>>>>>>>" + ex);
    	}finally{
    		in.close();
    		bis.close();
    		Logger.getRootLogger().info("UrlAbnorBolt_getStatConfigFiles  >>>>>>>>>>>>>>>> hmlen.size()=" +hmlen.size() );
    		Logger.getRootLogger().info("UrlAbnorBolt_getStatConfigFiles  >>>>>>>>>>>>>>>> hmpath.size()=" +hmpath.size() );
    		//Logger.getRootLogger().info("UrlAbnorBolt_getStatConfigFiles  >>>>>>>>>>>>>>>> hmpara.size()=" +hmpara.size() );
    		Logger.getRootLogger().info("UrlAbnorBolt_getStatConfigFiles  >>>>>>>>>>>>>>>> hmproba.size()=" +hmproba.size() );
//            IOUtils.closeStream(in); 
    	}
	}
	
	@Override
	public void execute(Tuple tuple) {
		//"dip","dport","sip","sport","gisstr","domainname","url"
		Date current = new Date();
		String strDate = new SimpleDateFormat("yyyyMMddHHmmss").format(current);
    	String dateStr_h = new SimpleDateFormat("yyyyMMddHH").format(current);
		String dateStr_d = new SimpleDateFormat("yyyyMMdd").format(current);
		String dateStr_hour = new SimpleDateFormat("HH:00").format(current);
		boolean isExec = false;
		String dip = tuple.getStringByField("dip");
		String dport = tuple.getStringByField("dport");
		String sip = tuple.getStringByField("sip");			
		String sport = tuple.getStringByField("sport");
		String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
		String domain = tuple.getStringByField("domainname");
		String url = tuple.getStringByField("url");
		
		if(domain.trim().isEmpty() || url.trim().isEmpty()) {
			return;
		}
		String[] dmarr = domainlist.split("\\|",-1);
		for(String dm : dmarr){
			if(domain.contains(dm)){
				isExec = true;
				break;
			}
		}
		if(!isExec) return;
		
		url = new String(Base64.decodeBase64(url)).toLowerCase();
		String[] sfarr = suffixname.split("\\|",-1);
		for(String sf : sfarr){
			if(url.contains("."+sf)) return;
		}
		
		
		String[] gisArray = gis.split("#",-1);
		if(gisArray.length!=7) {
			return;
		}
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6]; //如果省为空，精确到国家			
		
		try {	
			if(url.contains("://")){
				url = url.split("://",-1)[1];
			}
			
			
			//lenth model
			if(hmlen.containsKey(domain)){
				String[] arrlen = hmlen.get(domain).split("\\|",-1);
				if(arrlen.length!=2 || arrlen[0].trim().equals("") || arrlen[1].trim().equals("")) {
					Logger.getRootLogger().info(">>>>>>hmlen's length error <<<<<<");
					return;
				}
				int minlen = Integer.parseInt(arrlen[0].trim());
				int maxlen = Integer.parseInt(arrlen[1].trim());
				if(url.length() < minlen || url.length() > maxlen){					
					Logger.getRootLogger().info("execute length insertdata:url length|minlen|maxlen|tuple============="+ url.length()+"|"+ minlen+"|"+ maxlen+"|"+dip+"|"+dport+"|"+sip+"|"+sport+"|"+gis+"|"+domain+"|"+url);
					insertData(dip, dport, sip, sport, sourceAreaName, sourceGis, sourceAreaCountry, destAreaName, destGis, strDate);
					return;
				}
			}else{
				Logger.getRootLogger().info("execute length insertdata[new domain]:url length|tuple============="+ url.length()+"|"+dip+"|"+dport+"|"+sip+"|"+sport+"|"+gis+"|"+domain+"|"+url);
				insertData(dip, dport, sip, sport, sourceAreaName, sourceGis, sourceAreaCountry, destAreaName, destGis, strDate);
				return;
			}
			
			double critical = getCritical(domain,0);
			double proba = 1.0;
			//path model
			if(hmpath.containsKey(domain)){
				String[] urls = url.split("/",-1);
				int loop = urls.length > 3 ? 3 : urls.length-1;
				for (int i = 1; i < loop; i++) {
					if(urls[i-1].isEmpty() || urls[i].isEmpty()) {
						proba = 0.00;					
						break;
					}
					HashMap<String, Long> ls = hmpath.get(domain);
					String k1 = urls[i-1] + "/" + urls[i];
					if(ls.containsKey(urls[i-1]) && ls.containsKey(k1)){
						//Logger.getRootLogger().info("execute path model containsKey k-1|k|tuple>>>>>>>>>>>"+ls.get(k1)+"|"+ls.get(urls[i-1])+"|"+proba+"|"+dip+"|"+dport+"|"+sip+"|"+sport+"|"+gis+"|"+domain+"|"+url);
						proba = ((double)ls.get(k1) / (double)ls.get(urls[i-1])) * proba;
					}else{
						proba = 0.00;
						break;
					}
				}
			}else{
				proba = 0.00;
			}
			if(critical != 1.0 && proba <= critical){
				Logger.getRootLogger().info("execute path insertdata: domain|url|now critical|history critical|tuple============="+domain+"|"+url +"|"+ proba+"|"+ critical+"|"+dip+"|"+dport+"|"+sip+"|"+sport+"|"+gis);
				insertData(dip, dport, sip, sport, sourceAreaName, sourceGis, sourceAreaCountry, destAreaName, destGis, strDate);
				return;
			}
			
			//para model
//			if(url.contains("?") && hmpara.containsKey(domain)){
//				Map<String, String> mapRequest = CRequest.URLRequest(url);
//				proba = 1.0;
//				for(String strRequestKey: mapRequest.keySet()) {
//					String k1 = strRequestKey;
//					String k2 = CRequest.UrlPage(url)+"?"+strRequestKey;
//					HashMap<String, Long> ls = hmpara.get(domain);					
//					if(ls.containsKey(k1) && ls.containsKey(k2)){
//						proba = ((double)ls.get(k1) / (double)ls.get(k2)) * proba;
//						Logger.getRootLogger().info("execute para model containsKey>>>>>>>>>>>"+ls.get(k1)+"|"+ls.get(k2)+"|"+proba);
//					}else{
//						proba = 0.00;
//						break;
//					}
//				}				
//				critical = getCritical(domain,1);
//				if(critical != 1.0 && proba < critical){
//					Logger.getRootLogger().info("execute para insertdata:domain|url|now critical|history critical============="+domain+"|"+url +"|"+ proba+"|"+ critical);
//					insertData(dip, dport, sip, sport, sourceAreaName, sourceGis, sourceAreaCountry, destAreaName, destGis, strDate);
//					return;
//				}
//			}
			

			
			
    	
		} catch (Exception e) { 
			Logger.getRootLogger().error("UrlAbnorBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e);
		}finally{
			collector.emit(new Values(dip,sip,dport,sport,dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,sourceProvinceName, 1, dateStr_hour,"WEB"));
	    	collector.ack(tuple);
		}
	}
	
	public Double getCritical(String domain,int tag){
		String[] domainsp = domain.split("\\.",-1);
		String name = null;
		double critical = 1.0;
		for(int i=domainsp.length-1;i>0;i--){
			if(i == domainsp.length-1) name = domainsp[i];
			name = domainsp[i-1]+"." +name;
			if(!hmproba.containsKey(name)) continue;
			String[] prosp = hmproba.get(name).split("\\|",-1);
			if(tag == 0){
				critical = Double.parseDouble(prosp[0]);
				break;
			}else{
				critical = Double.parseDouble(prosp[1]);
				break;
			}	
		}
		if(critical == 1.00) 
			return 0.00;
		else
			return critical;
	}
	
	
	public void insertData(String dip,String dport,String sip,String sport,String sourceAreaName,String sourceGis,String sourceAreaCountry,String destAreaName,String destGis,String strDate){
		String rowkey = String.format("%s_%s_%s_%s_%s", dip, strDate,sip,dport,5);			
		String tbName = ApolloProperties.SDS_ABNORMAL_LOG;
		String cf = "cf";		
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver,logserver,brokerList);
		
		hbaseInstance.Add(tbName, rowkey, cf, "SOURCEIP", sip);
		hbaseInstance.Add(tbName, rowkey, cf, "DESTPORT", dport);							
		hbaseInstance.Add(tbName, rowkey, cf, "ACCESSTIME", strDate);
		hbaseInstance.Add(tbName, rowkey, cf, "ABRNORMAL", "5");
		hbaseInstance.Add(tbName, rowkey, cf, "DESTIP", dip);							
		hbaseInstance.Add(tbName, rowkey, cf, "SOURCEAREA", sourceAreaName);
		hbaseInstance.Add(tbName, rowkey, cf, "SOURCEGEO", sourceGis);
		hbaseInstance.Add(tbName, rowkey, cf, "SOURCECOUNTRY", sourceAreaCountry);
		hbaseInstance.Add(tbName, rowkey, cf, "DESTAREA", destAreaName);
		hbaseInstance.Add(tbName, rowkey, cf, "DESTGEO", destGis);							
		hbaseInstance.Add(tbName, rowkey, cf, "ATTNUM", "1");
		hbaseInstance.Add(tbName, rowkey, cf, "DESC", "WEB攻击");
		hbaseInstance.Add(tbName, rowkey, cf, "EVALUATE", "40");
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		suffixname =  stormConf.get("suffixname").toString();
		domainlist =  stormConf.get("domainlist").toString();
		
		try {
			conf.set("fs.hdfs.impl",org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());  
//			conf.set("fs.file.impl",org.apache.hadoop.fs.LocalFileSystem.class.getName());  
//	        conf.set("fs.defaultFS","hdfs://nameservice1");
			
			File f1 = new File("/etc/hadoop/conf/core-site.xml");
            if (f1.exists()) {
                conf.addResource(f1.toURI().toURL());
            }
            File f2 = new File("/etc/hadoop/conf/core-site.xml");
            if (f2.exists()) {
                conf.addResource(f2.toURI().toURL());
            }
            File f3 = new File("/etc/hadoop/conf/hdfs-site.xml");
            if (f3.exists()) {
                conf.addResource(f3.toURI().toURL());
            }
			getStatConfigFiles();
			
		} catch (MalformedURLException e1) {
			e1.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}


}
